package com.wf.mock.app.mina;

import java.util.Iterator;
import java.util.concurrent.DelayQueue;
import java.util.concurrent.Delayed;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

import lombok.Data;
import lombok.extern.log4j.Log4j;

import org.apache.commons.lang3.StringUtils;
import org.apache.mina.core.session.IoSession;

import com.google.common.cache.Cache;
import com.google.common.cache.CacheBuilder;
import com.wf.mock.app.mina.app.processor.IMessageProcessor;
import com.wf.mock.app.mina.app.processor.impl.MainMessageProcessorImpl;
import com.wf.mock.app.mina.dto.RawMessageDTO;

@Log4j
public enum RetrySendMsgManager {

	INSTANCE;

	private RetrySendMsgManager() {
		init();
	}

	private Cache<String, String> markIdCache = CacheBuilder.newBuilder().expireAfterWrite(10, TimeUnit.MINUTES)
			.build();

	private DelayQueue<Event> queue = new DelayQueue<Event>();

	private ExecutorService service = Executors.newFixedThreadPool(Runtime.getRuntime().availableProcessors());

	private IMessageProcessor mainMessageProcessor = MainMessageProcessorImpl.getInstance();

	private void init() {
		new Thread(new RetryTask(), "消息重试发送服务").start();
	}

	public void addEvent(RawMessageDTO msg, IoSession session) {

		String markId = msg.getMsgMarkId();
		// 确保一个消息只会被添加到重试队列中一次
		if (StringUtils.isNotBlank(markId) && markIdCache.getIfPresent(markId) == null) {
			queue.add(new Event(msg, session));
			markIdCache.put(markId, markId);
			log.info("添加重试发送事件成功:" + msg);
		}
	}

	/**
	 * 移除事件
	 * 
	 * @param markId
	 * @param userId
	 */
	public void removeEvent(String markId, String userId) {

		for (Iterator<Event> it = queue.iterator(); it.hasNext();) {
			Event event = it.next();
			if (markId.equals(event.getMsgMarkId()) && userId.equals(event.getUserId())) {
				it.remove();
				log.info("user::" + event.getUserId() + ", 移除重试发送事件成功:" + event.getMsg());
				return;
			}
		}
	}

	public void removeEvent(String markId) {

		for (Iterator<Event> it = queue.iterator(); it.hasNext();) {
			Event event = it.next();
			if (markId.equals(event.getMsgMarkId())) {
				it.remove();
				log.info("user::" + event.getUserId() + ",移除重试发送事件成功:" + event.getMsg());
				return;
			}
		}
	}

	/**
	 * 重试任务
	 * 
	 * @author wangyun
	 * 
	 */
	private final class RetryTask implements Runnable {
		@Override
		public void run() {
			while (true) {
				try {
					Event event = queue.poll(Long.MAX_VALUE, TimeUnit.DAYS);
					service.submit(buildSendTask(event));
				} catch (Exception e) {
					log.error(e);
				}
			}
		}

		private Runnable buildSendTask(final Event event) {
			return new Runnable() {
				@Override
				public void run() {
					mainMessageProcessor.process(event.getMsg(), event.getSession());
					if (event.decreRetryCount()) {
						queue.put(event);
					}
				}
			};
		}
	}

	/**
	 * 重试发送事件：用于包装原始消息，实现重发机制
	 * 
	 * @author wangyun
	 * 
	 */
	@Data
	class Event implements Delayed {

		/**
		 * 重试次数
		 */
		private final AtomicInteger retryCount = new AtomicInteger(3);

		/**
		 * 重试发送的时间
		 */
		private long retryTime;

		/**
		 * 原始消息内容
		 */
		private RawMessageDTO msg;

		private IoSession session;

		/**
		 * 重发事件构造器
		 * 
		 * @param msg
		 *            要重试发送的消息
		 */
		public Event(RawMessageDTO msg, IoSession session) {
			this.msg = msg;
			this.session = session;
			initRetryTime();
		}

		/**
		 * 减少重试次数：若减少后，重试次数还大于0，返回true,同时初始化下次重试发送时间； 若减少后，重试次数小于等于零,返回false;
		 * 
		 * @return 是否减少成功
		 */
		public boolean decreRetryCount() {

			if (retryCount.decrementAndGet() > 0) {
				initRetryTime();
				return true;
			} else {
				retryTime = Long.MAX_VALUE;
				return false;
			}

		}

		/**
		 * 
		 * 获取消息标示id
		 * 
		 * @return
		 */
		public String getMsgMarkId() {
			return msg.getMsgMarkId();
		}

		/**
		 * 
		 * 获取用户id
		 * 
		 * @return
		 */
		public String getUserId() {
			return session.getAttribute("userId").toString();
		}

		/**
		 * 初始化重试发送的时间
		 */
		private void initRetryTime() {
			retryTime = System.currentTimeMillis() + 10 * 1000;
		}

		@Override
		public int compareTo(Delayed o) {
			return Long.compare(this.getDelay(TimeUnit.MILLISECONDS), o.getDelay(TimeUnit.MILLISECONDS));
		}

		@Override
		public long getDelay(TimeUnit unit) {
			long diff = retryTime - System.currentTimeMillis();
			return unit.convert(diff, TimeUnit.MILLISECONDS);
		}

	}

}
